package com.liao.mr;

import com.liao.kv.key.ComDimension;
import com.liao.kv.value.CountDurationValue;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;

public class CountDurationRunner {
//    private Configuration conf = null;
//    @Override
//    public void setConf(Configuration configuration) {
//        this.conf = HBaseConfiguration.create(configuration);
//    }
//
//    @Override
//    public Configuration getConf() {
//        return this.conf;
//    }
//
//    @Override
//    public int run(String[] strings) throws Exception {
//        //1.获取配置和job对象
//        Configuration conf = this.getConf();
//        Job job = Job.getInstance(conf, "CALL_LOG_ANALYSIS");
//        //2.设置jar
//        job.setJarByClass(CountDurationRunner.class);
//        // 3.设置Mapper
//        this.setHBaseInputConfig(job);
//        //4.设置Reducer
//        job.setReducerClass(CountDurationReducer.class);
//        job.setOutputKeyClass(ComDimension.class);//
//        job.setOutputValueClass(CountDurationValue.class);
//        //5.设置输出类型&自定义的OutPutFormat
//        job.setOutputFormatClass(MySQLOutputFormat.class);
//        //6.提交
//        return job.waitForCompletion(true) ? 0 : 1;
//    }
//
//    public void setHBaseInputConfig(Job job) {
//        Configuration conf = job.getConfiguration();
//        HBaseAdmin admin = null;
//        try {
//            admin = new HBaseAdmin(conf);
//            //如果表不存在则直接返回，抛个异常也挺好
//            if(!admin.tableExists("ct:calllog")) throw new RuntimeException("Unable to find the specified table.");
//
//            Scan scan = new Scan();
//            scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME, Bytes.toBytes("ct:calllog"));
//            TableMapReduceUtil.initTableMapperJob("ct:calllog", scan, CountDurationMapper.class, ComDimension.class, Text.class, job, true);
//        } catch (IOException e) {
//            e.printStackTrace();
//        }finally {
//            if(admin != null) try {
//                admin.close();
//            } catch (IOException e) {
//                e.printStackTrace();
//            }
//        }
//    }
//
//    public static void main(String[] args) {
//        try {
//            int status = ToolRunner.run(new CountDurationRunner(), args);
//            System.exit(status);
//            if(status == 0){
//                System.out.println("运行成功");
//            }else {
//                System.out.println("运行失败");
//            }
//        } catch (Exception e) {
//            System.out.println("运行失败");
//            e.printStackTrace();
//        }
//    }

    public static void main(String[] args) throws Exception{
        Configuration conf = HBaseConfiguration.create();
        Job job = Job.getInstance(conf);

        job.setJarByClass(CountDurationRunner.class);

        Scan scan = new Scan();

        TableMapReduceUtil.initTableMapperJob(
                "ct:calllog",
                scan, CountDurationMapper.class,
                ComDimension.class, Text.class,job
        );

        job.setReducerClass(CountDurationReducer.class);
        job.setOutputKeyClass(ComDimension.class);
        job.setOutputValueClass(CountDurationValue.class);

        job.setOutputFormatClass(MySQLOutputFormat.class);

        boolean wait = job.waitForCompletion(true);
        System.exit(wait?0:1);
    }
}
